package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo11AggregateByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo11AggregateByKey")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 统计性别人数
    val stuLineRDD: RDD[String] = sc.textFile("Spark/data/students.txt")
    // 将数据变成KV格式 以性别作为Key 1作为Value
    val genderKVRDD: RDD[(String, Int)] = stuLineRDD.map(line => (line.split(",")(3), 1))

    genderKVRDD

      /**
       * zeroValue:U 自己定义的类型 初始值
       * seqOp: (U,Int) => U  ---> 相当于会对每个分区的数据进行处理，即在Map端进行预聚合
       * combOp: (U,U) => U  ---> 相当于会对每个Map任务最终的结果进行处理，即预聚合之后的结果再进行聚合
       *
       * zeroValue:Int
       * seqOp: (Int,Int) => Int
       * combOp: (Int,Int) => Int
       *
       */
      .aggregateByKey(0)(
        // 传入预聚合的方法 以及最终的聚合方法
        (i1, i2) => {
          i1 + i2
        } // 相当于combiner的处理逻辑
        , (u1, u2) => {
          u1 + u2
        } // 相当于reduce的处理逻辑
      ).foreach(println)

    /**
     * 统计每个班级的平均年龄：
     * 1、统计每个班级的年龄之和
     * 2、统计每个班级的学生数量
     */

    val clazzAgeKVRDD: RDD[(String, Int)] = stuLineRDD
      .map(line => (line.split(",")(4), line.split(",")(2).toInt))

    // 1、统计每个班级的年龄之和
    clazzAgeKVRDD
      .aggregateByKey(0)(
        (mapSumAge, age) => {
          mapSumAge + age
        }
        ,
        (map1SumAge, map2SumAge) => {
          map1SumAge + map2SumAge
        }
      ).foreach(println)

    // 2、统计每个班级的学生数量
    clazzAgeKVRDD
      .aggregateByKey(0)(
        (mapCnt, age) => {
          mapCnt + 1
        }
        ,
        (map1Cnt, map2Cnt) => {
          map1Cnt + map2Cnt
        }
      ).foreach(println)

    // 统计每个班级的平均年龄

    // 第一种方式：将上诉两个步骤算得的结果进行 join --> 计算班级的平均年龄

    // 第二种方式：将上诉两个步骤进行合并 --> 计算班级的平均年龄
    clazzAgeKVRDD
      // 相当于弥补了reduceByKey不能做avg的缺陷
      // 即实现了avg操作，又会在Map端进行预聚合
      .aggregateByKey((0, 0))(
        (t2, age) => {
          (t2._1 + age, t2._2 + 1)
        }
        ,
        (u1, u2) => {
          (u1._1 + u2._1, u1._2 + u2._2)
        }
      ).map(kv => s"${kv._1},${kv._2._1.toDouble / kv._2._2}")
      .foreach(println)

  }

}
